package spring.cloud.tasks.tasks_assign_manager.task_item_assign;

import com.google.gson.Gson;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import spring.cloud.tasks.tasks_assign_manager.domain.OnlineExecutor;
import spring.cloud.tasks.tasks_assign_manager.domain.TaskInstanceInfo;
import spring.cloud.tasks.tasks_assign_manager.domain.TaskItem;
import spring.cloud.tasks.tasks_assign_manager.domain.WaitedAssignTaskInfo;
import spring.cloud.tasks.tasks_assign_manager.node.Node;
import spring.cloud.tasks.tasks_assign_manager.schedule.CronExpressionContext;
import spring.cloud.tasks.tasks_assign_manager.schedule.CronExpressionContextManager;
import spring.cloud.tasks.tasks_assign_manager.schedule.TaskPreAssignTimePoint;

import java.nio.charset.StandardCharsets;
import java.util.*;

@Slf4j
@Setter
@Getter
public class TaskItemAssignUtils {


    /**
     * 获取该 nameSpace下的所有任务
     */
    public static List<String> getTaskIdList(CuratorFramework curatorFramework) throws Exception {
        if (curatorFramework.checkExists().forPath(Node.TASKS_NODE_PATH) == null) {
            curatorFramework.create().creatingParentsIfNeeded().forPath(Node.TASKS_NODE_PATH);
        }
        List<String> taskIdList = new ArrayList<>();
        List<String> $taskIdList = curatorFramework.getChildren().forPath(Node.TASKS_NODE_PATH);
        if ($taskIdList != null) {
            taskIdList.addAll($taskIdList);
        }
        return taskIdList;
    }

    /**
     * 获取该 nameSpace下的所有enable任务
     */
    public static List<String> getEnableTaskIdList(CuratorFramework curatorFramework, List<String> taskIdList) throws Exception {
        List<String> enableTaskIdList = new ArrayList<>();
        for (int i = 0; i < taskIdList.size(); i++) {
            String taskId = taskIdList.get(i);
            if (curatorFramework.checkExists().forPath(Node.getEnableTaskNodePath(taskId)) != null) {
                enableTaskIdList.add(taskId);
            }
        }
        return enableTaskIdList;
    }

    /**
     * 根据enableTaskIdList然后计算最近需要进行执行的任务。
     * 历史任务只要没有到达最大的timeoutDateTime那么则允许在任务分配列表上面，虽然当然对该任务不进行分配，但是有可能任务还在执行，这些任务即使执行失败也不修改执行计划。如果任务分配了还没有到执行的时间 那么可以进行重新分配。
     * 当然如果出现了节点上线或者下线这这种情况 只能等任务下次分配或者强制进行一次任务分配。
     * 除非强制要求下线的时候任务可以立即被分配（一天只执行一次 或者 有限的次数 那么允许重新分配 要不然很难被重新执行。任务级别fail over）。
     */

    public static List<TaskItem> getWaitedAssignTaskItemList(CuratorFramework curatorFramework, CronExpressionContextManager cronExpressionContextManager, List<String> enableTaskIdList) throws Exception {
        List<TaskItem> taskItemList = new ArrayList<>();
        //
        List<String> waitedAssignTaskIdList = new ArrayList<>();
        List<WaitedAssignTaskInfo> waitedAssignTaskInfoList = new ArrayList<>();
        Map<String, TaskPreAssignTimePoint> taskIdToTaskPreAssignTimePointMap = new HashMap<>();
        /**
         * 获取下次执行的计划时间。 真正进行任务项进行组装不是在这。
         */
        for (String enableTaskId : enableTaskIdList) {
            CronExpressionContext cronExpressionContext = cronExpressionContextManager.taskIdToCronExpressionContextMap.get(enableTaskId);
            if (cronExpressionContext == null) {
                continue;
            } else {
                WaitedAssignTaskInfo waitedAssignTaskInfo = getWaitedAssignTaskInfo(cronExpressionContextManager, enableTaskId, taskIdToTaskPreAssignTimePointMap.get(enableTaskId));
                if (waitedAssignTaskInfo != null) {
                    waitedAssignTaskIdList.add(enableTaskId);
                    waitedAssignTaskInfoList.add(waitedAssignTaskInfo);
                }
            }
        }
        for (WaitedAssignTaskInfo waitedAssignTaskInfo : waitedAssignTaskInfoList) {
            taskItemList.addAll(createTaskItemList(curatorFramework, waitedAssignTaskInfo));
        }
        return taskItemList;
    }

    /**
     * 判断当前时间点是否可以进行分配的任务
     */
    public static WaitedAssignTaskInfo getWaitedAssignTaskInfo(CronExpressionContextManager cronExpressionContextManager, String enableTaskId, TaskPreAssignTimePoint lastTaskPreAssignTimePoint) {
        CronExpressionContext cronExpressionContext = cronExpressionContextManager.taskIdToCronExpressionContextMap.get(enableTaskId);
        if (cronExpressionContext == null) {
            return null;
        }
        List<TaskPreAssignTimePoint> taskPreAssignTimePointList = cronExpressionContext.getTaskPreAssignTimePointList();
        if (taskPreAssignTimePointList.size() == 0) {
            return null;
        }
        //寻找可以预分配的时间节点，如果已经预分配了那么就不再进行分配了。
        //last预分配节点。
        int lastIndex = taskPreAssignTimePointList.size() - 1;
        Date now = new Date();
        if (lastTaskPreAssignTimePoint == null) {
            for (int i = 0; i <= lastIndex; i++) {//asc
                TaskPreAssignTimePoint taskPreAssignTimePoint = taskPreAssignTimePointList.get(i);
                Date nextFireDate = taskPreAssignTimePoint.getNextFireDate();
                Date nextPreAssignDate = taskPreAssignTimePoint.getNextPreAssignDate();
                if (now.getTime() > nextFireDate.getTime() + 1000l * 5) {
                    continue;
                } else {
                    if (now.getTime() < nextPreAssignDate.getTime()) {
                        return null;
                    } else {
                        cronExpressionContext.setLastPreAssignDate(nextPreAssignDate);
                        cronExpressionContext.setLastRealPreAssignDate(now);
                        //
                        WaitedAssignTaskInfo waitedAssignTaskInfo = new WaitedAssignTaskInfo();
                        waitedAssignTaskInfo.setTaskId(enableTaskId);
                        waitedAssignTaskInfo.setNextFireDate(nextFireDate);
                        waitedAssignTaskInfo.setTaskPreAssignTimePoint(taskPreAssignTimePoint);
                        return waitedAssignTaskInfo;
                    }
                }
            }
            return null;
        } else {
            for (int i = lastIndex; i >= 0; i--) {
                TaskPreAssignTimePoint taskPreAssignTimePoint = taskPreAssignTimePointList.get(i);
                if (taskPreAssignTimePoint.getNextFireDate().getTime() <= lastTaskPreAssignTimePoint.getNextFireDate().getTime()) {
                    continue;//有可能已经预分配了 距离下次分配还有一段时间
                } else {
                    //这代表可以预分配
                    Date nextFireDate = taskPreAssignTimePoint.getNextFireDate();
                    Date nextPreAssignDate = taskPreAssignTimePoint.getNextPreAssignDate();
                    if (now.getTime() > nextFireDate.getTime() + 1000l * 5) {
                        continue;
                    } else {
                        if (now.getTime() < nextPreAssignDate.getTime()) {
                            return null;
                        } else {
                            cronExpressionContext.setLastPreAssignDate(nextPreAssignDate);
                            cronExpressionContext.setLastRealPreAssignDate(now);
                            //
                            WaitedAssignTaskInfo waitedAssignTaskInfo = new WaitedAssignTaskInfo();
                            waitedAssignTaskInfo.setTaskId(enableTaskId);
                            waitedAssignTaskInfo.setNextFireDate(nextFireDate);
                            waitedAssignTaskInfo.setTaskPreAssignTimePoint(taskPreAssignTimePoint);
                            return waitedAssignTaskInfo;
                        }
                    }
                }
            }
            return null;
        }

    }


    public static List<OnlineExecutor> copyExecutorList(List<OnlineExecutor> onlineExecutorList) {
        List<OnlineExecutor> $onlineExecutorList = new ArrayList<>();
        for (OnlineExecutor onlineExecutor : onlineExecutorList) {
            OnlineExecutor $onlineExecutor = new OnlineExecutor();
            $onlineExecutor.setExecutorId(onlineExecutor.getExecutorId());
            $onlineExecutor.setIp(onlineExecutor.getIp());
            $onlineExecutor.setOwnTraffic(onlineExecutor.isOwnTraffic());
            $onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel());
            if (onlineExecutor.getSupportedTaskIdList() != null) {
                $onlineExecutor.setSupportedTaskIdList(new ArrayList<String>());
                for (String taskId : onlineExecutor.getSupportedTaskIdList()) {
                    $onlineExecutor.getSupportedTaskIdList().add(taskId);
                }
            }
            if (onlineExecutor.getAssignedTaskItemList() != null) {
                $onlineExecutor.setAssignedTaskItemList(new ArrayList<TaskItem>());
                for (TaskItem taskItem : onlineExecutor.getAssignedTaskItemList()) {
                    TaskItem $taskItem = new TaskItem();
                    $taskItem.copy(taskItem);
                    $onlineExecutor.getAssignedTaskItemList().add($taskItem);
                }
            }
            $onlineExecutorList.add($onlineExecutor);
        }
        return $onlineExecutorList;
    }

    public static List<OnlineExecutor> getOwnTrafficOnlineExecutorList(List<OnlineExecutor> onlineExecutorList) {
        List<OnlineExecutor> ownTrafficOnlineExecutorList = new ArrayList<>();
        for (OnlineExecutor executor : onlineExecutorList) {
            if (executor.isOwnTraffic()) {
                ownTrafficOnlineExecutorList.add(executor);
            }
        }
        return ownTrafficOnlineExecutorList;
    }

    /**
     * 按照loadLevel降序排序，如果loadLevel相同，按照任务名降序排序
     */
    public static void sortTaskItemList(List<TaskItem> taskItemList) {
        Collections.sort(taskItemList, new Comparator<TaskItem>() {
            @Override
            public int compare(TaskItem o1, TaskItem o2) {
                int sort = o2.getLoadLevel() - o1.getLoadLevel();
                return sort != 0 ? sort : o2.getTaskId().compareTo(o1.getTaskId());
            }
        });
    }

    public static boolean preferExecutorIdListIsConfigured(CuratorFramework curatorFramework, String taskId) throws Exception {
        if (curatorFramework.checkExists().forPath(Node.getTaskConfigPreferListNodePath(taskId)) != null) {
            byte[] bytes = curatorFramework.getData().forPath(Node.getTaskConfigPreferListNodePath(taskId));
            if (bytes != null) {
                return new String(bytes, StandardCharsets.UTF_8.name()).trim().length() > 0;
            }
        }
        return false;
    }

    /**
     * 获取配置态的preferList，即使配置的executor不存在，也会返回。 特别的是，对于docker task，如果存在，才去解析出executor列表。
     */
    public static List<String> getPreferExecutorIdListConfigured(CuratorFramework curatorFramework, String taskId) throws Exception {
        List<String> preferExecutorIdList = new ArrayList<>();
        if (curatorFramework.checkExists().forPath(Node.getTaskConfigPreferListNodePath(taskId)) != null) {
            byte[] bytes = curatorFramework.getData().forPath(Node.getTaskConfigPreferListNodePath(taskId));
            if (bytes != null) {
                String[] preferExecutorIds = new String(bytes, StandardCharsets.UTF_8.name()).split(",");
                List<String> executorIdList = getExecutorIdList(curatorFramework);
                for (String preferExecutorId : preferExecutorIds) {//min check way
                    if (executorIdList.contains(preferExecutorId) && !preferExecutorIdList.contains(preferExecutorId)) {
                        preferExecutorIdList.add(preferExecutorId);
                    }
                }
            }
        }
        return preferExecutorIdList;
    }

    public static List<String> getExecutorIdList(CuratorFramework curatorFramework) throws Exception {
        List<String> executorIdList = new ArrayList<>();
        if (curatorFramework.checkExists().forPath(Node.getExecutorsNodePath()) != null) {
            List<String> $executorIdList = curatorFramework.getChildren().forPath(Node.getExecutorsNodePath());
            if ($executorIdList != null) {
                executorIdList.addAll($executorIdList);
            }
        }
        return executorIdList;
    }

    public static List<OnlineExecutor> getSupportedTaskIdOwnTrafficOnlineExecutorList(List<OnlineExecutor> onlineExecutorList, String taskId) {
        List<OnlineExecutor> $executorList = new ArrayList<>();
        for (int i = 0; i < onlineExecutorList.size(); i++) {
            OnlineExecutor onlineExecutor = onlineExecutorList.get(i);
            List<String> taskIdList = onlineExecutor.getSupportedTaskIdList();
            if (taskIdList != null && taskIdList.contains(taskId)) {
                $executorList.add(onlineExecutor);
            }
        }
        return $executorList;
    }

    /**
     * 平衡摘取算法-非多余权重保留模式。保证单个任务均衡不影响所有执行节点。
     *
     * @param onlineExecutorList
     * @param taskItemList
     */
    public static void balancePickUp(List<OnlineExecutor> onlineExecutorList, String taskId, List<TaskItem> taskItemList) {
        // 计算出executor对该任务的平均负荷
        int totalLoadLevel = 0;
        for (TaskItem taskItem : taskItemList) {
            totalLoadLevel += taskItem.getLoadLevel();
        }
        for (OnlineExecutor onlineExecutor : onlineExecutorList) {
            for (TaskItem taskItem : onlineExecutor.getAssignedTaskItemList()) {
                if (taskId.equals(taskItem.getTaskId())) {
                    totalLoadLevel += taskItem.getLoadLevel();
                }
            }
        }
        int averageTotalLoadLevelAtPerExecutor = totalLoadLevel / onlineExecutorList.size();
        //
        // 摘取executor超出平均负荷的分片
        for (OnlineExecutor onlineExecutor : onlineExecutorList) {
            int taskTotalLoadLevel = 0;
            Iterator<TaskItem> iterator = onlineExecutor.getAssignedTaskItemList().iterator();
            while (iterator.hasNext()) {
                TaskItem taskItem = iterator.next();
                if (taskId.equals(taskItem.getTaskId())) {
                    //超过平均权重则不保留。
                    if (averageTotalLoadLevelAtPerExecutor == 0 || taskTotalLoadLevel + taskItem.getLoadLevel() > averageTotalLoadLevelAtPerExecutor) {
                        iterator.remove();
                        onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getLoadLevel());
                        //
                        taskItemList.add(taskItem);
                    } else {
                        //不超过平均权重则保留。
                        taskTotalLoadLevel += taskItem.getLoadLevel();
                    }
                }
            }
        }
    }

    public static void putBackWithPreferExecutorIdListConfigured(List<TaskItem> pickUpEdTaskItemList,//
                                                                 Map<String, List<OnlineExecutor>> taskIdToOwnTrafficOnlineExecutorListMap,//
                                                                 Map<String, Boolean> taskIdToIsConfiguredPreferExecutorIdListMap,//
                                                                 Map<String, List<String>> taskIdToConfiguredPreferExecutorIdListMap) {//
        Iterator<TaskItem> iterator = pickUpEdTaskItemList.iterator();
        while (iterator.hasNext()) {
            TaskItem taskItem = iterator.next();
            String taskId = taskItem.getTaskId();
            // preferList为空不能作为判断是否配置preferList的依据，比如说配置了容器资源，但是全部下线了。
            if (taskIdToIsConfiguredPreferExecutorIdListMap.get(taskId)) {
                List<String> preferExecutorIdListConfigured = taskIdToConfiguredPreferExecutorIdListMap.get(taskId);
                List<OnlineExecutor> preferOwnTrafficOnlineExecutorList = getPreferOwnTrafficOnlineExecutorList(taskIdToOwnTrafficOnlineExecutorListMap, taskId, preferExecutorIdListConfigured);
                // 如果preferList的Executor都offline，则放回到全部online的Executor中某一个。如果是这种情况，则后续再操作，避免不均衡的情况
                // 如果存在preferExecutor，择优放回
                if (preferOwnTrafficOnlineExecutorList != null && !preferOwnTrafficOnlineExecutorList.isEmpty()) {
                    OnlineExecutor onlineExecutor = getOnlineExecutorWithMinTaskTotalLoadLevelOnOnlineExecutor(preferOwnTrafficOnlineExecutorList, taskId);
                    addTaskItemToOnlineExecutor(taskItem, onlineExecutor);
                    iterator.remove();
                } else {
                    // 如果不存在preferExecutor
                    // 等到后续（在第3步）进行放回操作，避免不均衡的情况
                }
            }
        }
    }

    public static List<OnlineExecutor> getPreferOwnTrafficOnlineExecutorList(Map<String, List<OnlineExecutor>> taskIdToOwnTrafficOnlineExecutorListMap, String taskId, List<String> preferExecutorIdListConfigured) {
        List<OnlineExecutor> preferOnlineExecutorList = new ArrayList<>();
        List<OnlineExecutor> ownTrafficOnlineExecutorList = taskIdToOwnTrafficOnlineExecutorListMap.get(taskId);
        for (int i = 0; i < ownTrafficOnlineExecutorList.size(); i++) {
            OnlineExecutor onlineExecutor = ownTrafficOnlineExecutorList.get(i);
            if (preferExecutorIdListConfigured.contains(onlineExecutor.getExecutorId())) {
                preferOnlineExecutorList.add(onlineExecutor);
            }
        }
        return preferOnlineExecutorList;
    }

    public static void putBackWithoutPreferExecutorIdListConfigured(List<TaskItem> pickUpEdTaskItemList, Map<String, List<OnlineExecutor>> taskIdToOwnTrafficOnlineExecutorListMap) {
        Iterator<TaskItem> iterator = pickUpEdTaskItemList.iterator();
        while (iterator.hasNext()) {
            TaskItem taskItem = iterator.next();
            List<OnlineExecutor> taskOwnTrafficOnlineExecutorList = taskIdToOwnTrafficOnlineExecutorListMap.get(taskItem.getTaskId());//重点是这段代码 如果没有执行器怎么办???
            if (taskOwnTrafficOnlineExecutorList != null && !taskOwnTrafficOnlineExecutorList.isEmpty()) {
                OnlineExecutor onlineExecutor = getOnlineExecutorWithMinTaskTotalLoadLevelOnOnlineExecutor(taskOwnTrafficOnlineExecutorList, taskItem.getTaskId());
                addTaskItemToOnlineExecutor(taskItem, onlineExecutor);
                iterator.remove();
            }

        }
    }

    /**
     * 获取该任务负荷最小的executor，如果相同，那么取所有任务负荷最小的executor
     */
    public static OnlineExecutor getOnlineExecutorWithMinTaskTotalLoadLevelOnOnlineExecutor(List<OnlineExecutor> taskOwnTrafficOnlineExecutorList, String taskId) {
        OnlineExecutor minTaskTotalLoadLevelOnlineExecutor = null;
        int minTaskTotalLoadLevelOnOnlineExecutor = 0;
        for (int i = 0; i < taskOwnTrafficOnlineExecutorList.size(); i++) {
            OnlineExecutor onlineExecutor = taskOwnTrafficOnlineExecutorList.get(i);
            int taskTotalLoadLevelOnExecutor = getTaskTotalLoadLevelOnExecutor(onlineExecutor, taskId);
            if (minTaskTotalLoadLevelOnlineExecutor == null || minTaskTotalLoadLevelOnOnlineExecutor > taskTotalLoadLevelOnExecutor || minTaskTotalLoadLevelOnOnlineExecutor == taskTotalLoadLevelOnExecutor && minTaskTotalLoadLevelOnlineExecutor.getTotalLoadLevel() > onlineExecutor.getTotalLoadLevel()) {
                minTaskTotalLoadLevelOnlineExecutor = onlineExecutor;
                minTaskTotalLoadLevelOnOnlineExecutor = taskTotalLoadLevelOnExecutor;
            }
        }
        return minTaskTotalLoadLevelOnlineExecutor;
    }

    /**
     * 计算指定任务在特定executor上的负荷
     */
    public static int getTaskTotalLoadLevelOnExecutor(OnlineExecutor onlineExecutor, String taskId) {
        int taskTotalLoadLevel = 0;
        List<TaskItem> taskItemList = onlineExecutor.getAssignedTaskItemList();
        for (int i = 0; i < taskItemList.size(); i++) {
            TaskItem taskItem = taskItemList.get(i);
            if (taskId.equals(taskItem.getTaskId())) {
                taskTotalLoadLevel += taskItem.getLoadLevel();
            }
        }
        return taskTotalLoadLevel;
    }

    public static void addTaskItemToOnlineExecutor(TaskItem taskItem, OnlineExecutor onlineExecutor) {
        if (onlineExecutor != null) {
            if (isIn(onlineExecutor.getAssignedTaskItemList(), taskItem)) {
                log.error("The shard({}-{}) is running in the executor of {}, cannot be put again", taskItem.getTaskId(), taskItem.getTaskItemIndex(), onlineExecutor.getExecutorId());
            } else {
                onlineExecutor.getAssignedTaskItemList().add(taskItem);
                onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() + taskItem.getLoadLevel());
            }
        } else {
            log.info("No executor to take over the shard: {}-{}", taskItem.getTaskId(), taskItem.getTaskItemIndex());
        }
    }

    public static boolean isIn(List<TaskItem> taskItemList, TaskItem taskItem) {
        for (int i = 0; i < taskItemList.size(); i++) {
            TaskItem $taskItem = taskItemList.get(i);
            if ($taskItem.getTaskId().equals(taskItem.getTaskId()) && $taskItem.getTaskItemIndex() == taskItem.getTaskItemIndex()) {
                return true;
            }
        }
        return false;
    }

    public static boolean executorListContentIsChanged(List<OnlineExecutor> oldOnlineExecutorList, List<OnlineExecutor> newOnlineExecutorList) {
        return !(new Gson()).toJson(oldOnlineExecutorList).equals((new Gson()).toJson((newOnlineExecutorList)));
    }

    public static boolean executorIsOwnTraffic(CuratorFramework curatorFramework, String executorId) throws Exception {
        byte[] bytes = curatorFramework.getData().forPath(Node.getExecutorOwnTrafficNodePath(executorId));//Y有流量 N没有流量  默认是Y
        return bytes != null && Boolean.parseBoolean(new String(bytes));
    }

    /**
     * 创建全新的任务项item list
     */
    public static List<TaskItem> createTaskItemList(CuratorFramework curatorFramework, WaitedAssignTaskInfo waitedAssignTaskInfo) throws Exception {
        List<TaskItem> taskItemList = new ArrayList<>();
        String taskType = waitedAssignTaskInfo.getTaskType();
        //如果是item task or message task那么只有root instance task,基本上就只有root实例。
        if (TaskItem.TASK_TYPE_ITEM_TASK.equals(taskType) || TaskItem.TASK_TYPE_MESSAGE_TASK.equals(taskType)) {
            String taskInstanceId = "root";
            // 新建shardingTotalCount数量的分片
            taskItemList.addAll(createTaskItemList(waitedAssignTaskInfo, taskInstanceId));
            return taskItemList;
        } else if (TaskItem.TASK_TYPE_BATCH_TASK.equals(taskType)) {
            //batch task允许慢慢处理。
            //如果是batch task 有N+2个任务分片,1任务分配+N执行器+1状态收集(任务分配和状态收集尽可能不要进行随意切换执行器)。
            //如果batch task实例任务分配没有完成 那么只有一个任务分片处理。分配完成后有N+1进行处理。
            //所以针对batch task 需要一种汇报机制。任务分配完成后30秒内进行任务的执行调度，允许不是马上执行。
            List<TaskInstanceInfo> taskInstanceInfoList = getTaskInstanceInfoList(waitedAssignTaskInfo.getTaskId());
            for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                String taskInstanceId = taskInstanceInfo.getTaskInstanceId();
                if (TaskInstanceInfo.status_init.equals(taskInstanceInfo.getStatus())) {
                    TaskItem taskItem = new TaskItem();
                    taskItem.setTaskId(waitedAssignTaskInfo.getTaskId());
                    taskItem.setTaskType(taskType);
                    taskItem.setTaskInstanceId(taskInstanceId);
                    taskItem.setTaskItemIndex(-1);//进行数据拆分
                    taskItem.setLoadLevel(waitedAssignTaskInfo.getLoadLevel());
                    taskItem.setNextFireDate(waitedAssignTaskInfo.getNextFireDate());
                    taskItem.setTimeoutDateTime(waitedAssignTaskInfo.getTimeoutDateTime());
                    taskItem.setLockDateTime(waitedAssignTaskInfo.getLockDateTime());
                    taskItemList.add(taskItem);
                    taskInstanceInfo.setStatus(TaskInstanceInfo.status_data_splitting);
                } else if (TaskInstanceInfo.status_data_splitting.equals(taskInstanceInfo.getStatus())) {

                } else if (TaskInstanceInfo.status_data_split_success.equals(taskInstanceInfo.getStatus())) {
                    TaskItem taskItem = new TaskItem();
                    taskItem.setTaskId(waitedAssignTaskInfo.getTaskId());
                    taskItem.setTaskType(taskType);
                    taskItem.setTaskInstanceId(taskInstanceId);
                    taskItem.setTaskItemIndex(-11);//任务状态收集
                    taskItem.setLoadLevel(waitedAssignTaskInfo.getLoadLevel());
                    taskItem.setNextFireDate(waitedAssignTaskInfo.getNextFireDate());
                    taskItem.setTimeoutDateTime(waitedAssignTaskInfo.getTimeoutDateTime());
                    taskItem.setLockDateTime(waitedAssignTaskInfo.getLockDateTime());
                    taskItemList.add(taskItem);
                    //
                    taskItemList.addAll(createTaskItemList(waitedAssignTaskInfo, taskInstanceId));
                } else if (TaskInstanceInfo.status_data_split_fail.equals(taskInstanceInfo.getStatus())) {

                }
            }
            return taskItemList;
        } else {
            throw new IllegalStateException();
        }

    }


    public static List<TaskInstanceInfo> getTaskInstanceInfoList(String taskId) {
        //获得当时能够运行的任务分片信息
        List<TaskInstanceInfo> taskInstanceInfoList = new ArrayList<>();
        TaskInstanceInfo taskInstanceInfo = new TaskInstanceInfo();
        taskInstanceInfo.setTaskId(taskId);
        taskInstanceInfo.setTaskInstanceId(UUID.randomUUID().toString());
        taskInstanceInfoList.add(taskInstanceInfo);
        return taskInstanceInfoList;//
    }

    public static int getTotalTaskItemCount(CuratorFramework curatorFramework, String taskId) throws Exception {
        String totalTaskItemCountNodePath = Node.getTaskConfigTotalTaskItemCountNodePath(taskId);
        if (curatorFramework.checkExists().forPath(totalTaskItemCountNodePath) != null) {
            byte[] bytes = curatorFramework.getData().forPath(totalTaskItemCountNodePath);
            if (bytes != null) {
                try {
                    int totalTaskItemCount = Integer.parseInt(new String(bytes, StandardCharsets.UTF_8.name()));
                    return totalTaskItemCount;
                } catch (NumberFormatException e) {
                    log.error("parse shardingTotalCount error, will use the default value", e);
                    throw new RuntimeException(e);
                }
            } else {
                throw new RuntimeException();
            }
        } else {
            throw new RuntimeException();
        }
    }

    public static int getLoadLevel(CuratorFramework curatorFramework, String taskId) throws Exception {
        String taskConfigLoadLevelNodePath = Node.getTaskConfigLoadLevelNodePath(taskId);
        if (curatorFramework.checkExists().forPath(taskConfigLoadLevelNodePath) != null) {
            byte[] bytes = curatorFramework.getData().forPath(taskConfigLoadLevelNodePath);
            if (bytes != null) {
                int loadLevel = Integer.parseInt(new String(bytes, StandardCharsets.UTF_8.name()));
                return loadLevel;
            } else {
                throw new RuntimeException();
            }
        } else {
            throw new RuntimeException();
        }
    }

    /**
     * 分片项只能是1-N,不能以0开头
     */
    public static List<TaskItem> createTaskItemList(WaitedAssignTaskInfo waitedAssignTaskInfo, String taskInstanceId) {
        List<TaskItem> taskItemList = new ArrayList<>();
        String taskId = waitedAssignTaskInfo.getTaskId();
        String taskType = waitedAssignTaskInfo.getTaskType();
        int totalTaskItemCount = waitedAssignTaskInfo.getTotalTaskItemCount();
        int loadLevel = waitedAssignTaskInfo.getLoadLevel();
        Date nextFireDate = waitedAssignTaskInfo.getNextFireDate();
        Date timeoutDateTime = waitedAssignTaskInfo.getTimeoutDateTime();
        Date lockDateTime = waitedAssignTaskInfo.getLockDateTime();
        for (int taskItemIndex = 1; taskItemIndex <= totalTaskItemCount; taskItemIndex++) {
            TaskItem taskItem = new TaskItem();
            taskItem.setTaskId(taskId);
            taskItem.setTaskType(taskType);
            taskItem.setTaskInstanceId(taskInstanceId);
            taskItem.setTaskItemIndex(taskItemIndex);
            taskItem.setLoadLevel(loadLevel);
            taskItem.setNextFireDate(nextFireDate);
            taskItem.setTimeoutDateTime(timeoutDateTime);
            taskItem.setLockDateTime(lockDateTime);
            taskItemList.add(taskItem);
        }
        return taskItemList;
    }

    /**
     * 刷新onlineExecutorList taskIdList
     */
    public static boolean refreshOnlineExecutorSupportedTaskIdList(CuratorFramework curatorFramework, List<OnlineExecutor> onlineExecutorList, String taskId) throws Exception {
        boolean fixed = false;
        for (int i = 0; i < onlineExecutorList.size(); i++) {
            OnlineExecutor onlineExecutor = onlineExecutorList.get(i);
            if (onlineExecutor.getSupportedTaskIdList() == null) {
                onlineExecutor.setSupportedTaskIdList(new ArrayList<String>());
            }
            List<String> taskIdList = onlineExecutor.getSupportedTaskIdList();
            String taskExecutorStatusNodePath = Node.getTaskExecutorStatusNodePath(taskId, onlineExecutor.getExecutorId());
            if (curatorFramework.checkExists().forPath(taskExecutorStatusNodePath) != null) {
                if (!taskIdList.contains(taskId)) {
                    taskIdList.add(taskId);
                    fixed = true;
                }
            } else {
                if (taskIdList.contains(taskId)) {
                    taskIdList.remove(taskId);
                    fixed = true;
                }
            }
        }
        return fixed;
    }

    /**
     * 移除特定任务的shard
     */
    public static List<TaskItem> removeTaskItemListOfTask(List<OnlineExecutor> ownTrafficOnlineExecutorList, String taskId) {
        List<TaskItem> removedTaskItemList = new ArrayList<>();
        for (int i = 0; i < ownTrafficOnlineExecutorList.size(); i++) {
            OnlineExecutor onlineExecutor = ownTrafficOnlineExecutorList.get(i);
            Iterator<TaskItem> iterator = onlineExecutor.getAssignedTaskItemList().iterator();
            while (iterator.hasNext()) {
                TaskItem taskItem = iterator.next();
                if (taskId.equals(taskItem.getTaskId())) {
                    onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getLoadLevel());
                    iterator.remove();
                    removedTaskItemList.add(taskItem);
                }
            }
        }
        return removedTaskItemList;
    }

    /**
     * @return 可能返回为空
     * @throws Exception
     */
    public static String getIp(CuratorFramework curatorFramework, String executorId) throws Exception {
        String ip = null;
        String executorIpNodePath = Node.getExecutorIpNodePath(executorId);
        if (curatorFramework.checkExists().forPath(Node.getExecutorIpNodePath(executorId)) != null) {
            byte[] bytes = curatorFramework.getData().forPath(executorIpNodePath);
            if (bytes != null) {
                ip = new String(bytes, StandardCharsets.UTF_8);
            }
        }
        return ip;
    }

    public static boolean hasTaskItemRunning(String taskId, List<OnlineExecutor> onlineOwnTrafficExecutorList) {
        for (int i = 0; i < onlineOwnTrafficExecutorList.size(); i++) {
            List<TaskItem> taskItemList = onlineOwnTrafficExecutorList.get(i).getAssignedTaskItemList();
            for (int j = 0; j < taskItemList.size(); j++) {
                if (taskItemList.get(j).getTaskId().equals(taskId)) {
                    return true;
                }
            }
        }
        return false;
    }

    public static List<OnlineExecutor> getOnlineExecutorListInPreferOrDisPreferExecutorIdListConfigured(
            String taskId,//
            List<String> preferExecutorIdListConfigured,//
            List<OnlineExecutor> onlineOwnTrafficExecutorList,//
            boolean contains) {//
        List<OnlineExecutor> onlineExecutorList = new ArrayList<>();
        for (OnlineExecutor onlineExecutor : onlineOwnTrafficExecutorList) {
            if (onlineExecutor.getSupportedTaskIdList().contains(taskId) && contains == preferExecutorIdListConfigured.contains(onlineExecutor.getExecutorId())) {
                onlineExecutorList.add(onlineExecutor);
            }
        }
        return onlineExecutorList;
    }

    public static List<TaskItem> pickTaskItemListInDisPreferExecutorIdList(
            String taskId,//
            List<String> preferExecutorIdListConfigured, List<OnlineExecutor> onlineOwnTrafficExecutorList) {
        List<TaskItem> taskItemList = new ArrayList<>();
        for (int i = 0; i < onlineOwnTrafficExecutorList.size(); i++) {
            OnlineExecutor onlineExecutor = onlineOwnTrafficExecutorList.get(i);
            if (!preferExecutorIdListConfigured.contains(onlineExecutor.getExecutorId())) {
                Iterator<TaskItem> iterator = onlineExecutor.getAssignedTaskItemList().iterator();
                while (iterator.hasNext()) {
                    TaskItem taskItem = iterator.next();
                    if (taskItem.getTaskId().equals(taskId)) {
                        iterator.remove();
                        onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getLoadLevel());
                        //
                        taskItemList.add(taskItem);
                    }
                }
            }
        }
        return taskItemList;
    }

    public static boolean taskItemIsAllRunningInDisPreferOnlineExecutorList(String executorId, String taskId, List<String> preferExecutorIdListConfigured, List<OnlineExecutor> onlineOwnTrafficExecutorList) {
        for (int i = 0; i < onlineOwnTrafficExecutorList.size(); i++) {
            OnlineExecutor onlineExecutor = onlineOwnTrafficExecutorList.get(i);
            if (preferExecutorIdListConfigured.contains(executorId)) {
                List<TaskItem> taskItemList = onlineExecutor.getAssignedTaskItemList();
                for (int j = 0; j < taskItemList.size(); j++) {
                    if (taskItemList.get(j).getTaskId().equals(taskId)) {
                        return false;
                    }
                }
            }
        }
        return true;
    }

    public static List<OnlineExecutor> getOnlineExecutorList(List<OnlineExecutor> onlineOwnTrafficExecutorList, String taskId) {
        List<OnlineExecutor> onlineExecutorList = new ArrayList<>();
        for (OnlineExecutor onlineExecutor : onlineOwnTrafficExecutorList) {
            if (onlineExecutor.getSupportedTaskIdList().contains(taskId)) {
                onlineExecutorList.add(onlineExecutor);
            }
        }
        return onlineExecutorList;
    }
}
